package client_agent

import (
	"fmt"
	"lark/conf"
	"lark/consts"
	"lark/proto"
	"lark/util"
	"net"
	"sync/atomic"

	"context"

	"sync"
	"time"

	gproto "github.com/golang/protobuf/proto"
	"github.com/pkg/errors"
)

type server struct {
	sync.Mutex
	id        string
	raddr     string
	msgid     uint64
	parser    *proto.LarkProtoV2
	status    uint32
	workerNum int
	reqMap    map[uint64]*sessionInMsg
}

func newServer(addr string, workerNum int) *server {
	return &server{
		id:        "server-" + addr,
		raddr:     addr,
		status:    newFlag,
		workerNum: workerNum,
		reqMap:    make(map[uint64]*sessionInMsg),
	}
}

func (s *server) run() {
	s.Lock()
	defer s.Unlock()

	if s.status == newFlag {
		if err := s.handshake(); err != nil {
			logger.Errorf("%+v", err)

		} else {
			s.status = runningFlag
		}
	}
	return
}

func (s *server) handshake() error {
	conn, err := net.DialTimeout("tcp", s.raddr, conf.DefaultNetDialTimeout())
	if err != nil {
		return errors.New(fmt.Sprintf("[%s] %s", s.id, err))
	}

	//设置 keepalive 参数可以在中间有 VPN 等网络设备的时候，增加一个保活手段
	conn.(*net.TCPConn).SetKeepAlive(true)
	conn.(*net.TCPConn).SetKeepAlivePeriod(time.Minute)

	s.parser = proto.NewlarkProtoV2(conn, conf.ClientAgentReadRemoteTimeout(), conf.ClientAgentWriteRemoteTimeout())

	//打包握手数据
	b, err := gproto.Marshal(&proto.Handshake{Version: conf.ProtoVersion1})
	if err != nil {
		return errors.New(fmt.Sprintf("[%s] marshal handshake failed. %s", s.id, err))
	}

	//发送握手数据
	err = s.parser.Write(consts.Lcp2lspTagHandshakeReq, b)
	if err != nil {
		return errors.Errorf("[%s] send handshake failed. %s", s.id, err)
	}

	//接收握手返回
	tag, payload, err := s.parser.ReadTimeout(conf.DefaultNetReadTimeout())
	if err != nil {
		return errors.Errorf("[%s] receive handshake failed. %s", s.id, err)
	}

	//判断 tag
	if tag != consts.Lsp2lcpTagHandshakeResp {
		return errors.Errorf("[%s] receive wrong handshake tag. want:%d, get:%d", s.id, consts.Lsp2lcpTagHandshakeResp, tag)
	}

	//解析握手返回
	var msg = new(proto.Handshake)
	err = gproto.Unmarshal(proto.SplitPayload(payload)[0], msg)
	if err != nil {
		return errors.Errorf("[%s] unmarhsal handshake failed. %s", s.id, err)
	}

	//判断使用哪一种协议版本
	//方便以后针对不同的版本使用不同的逻辑
	switch msg.Version {
	case conf.ProtoVersion1:
		logger.Debugf("[%s] handshake success. use proto version:%d", s.id, conf.ProtoVersion1)
		go s.readLoop()

	default:
		return errors.Errorf("[%s] can't support proto version:%d", s.id, msg.Version)
	}

	return nil
}

func (s *server) write(msg *sessionInMsg, ctx context.Context) error {
	s.Lock()
	defer s.Unlock()

	if s.status != runningFlag {
		s.descMsg(msg.msgid)
		return errors.Errorf("server closed, ip:%s", s.raddr)
	}

	var timeout time.Duration
	line, ok := ctx.Deadline()
	if ok {
		timeout = line.Sub(time.Now())
		if timeout <= 0 {
			return errors.Errorf("timeout before write data")
		}
	} else {
		timeout = conf.ClientAgentWriteRemoteTimeout()
	}

	err := s.parser.WriteWithTimeout(timeout, consts.Lcp2lspTagTaskReq, msg.data)
	if err != nil {
		s.descMsg(msg.msgid)
		return errors.Errorf("send data fail. ip:%s, error:%s", s.id, err.Error())
	}

	s.reqMap[msg.msgid] = msg
	return nil
}

func (s *server) incrMsg(msgid uint64) {
	s.Lock()
	s.reqMap[msgid] = nil
	s.Unlock()
}

func (s *server) descMsg(msgid uint64) {
	delete(s.reqMap, msgid)
}

func (s *server) readLoop() {
	defer util.PrintPanicStack()
	defer s.close()

	for {
		tag, payload, err := s.parser.ReadTimeout(conf.ClientAgentReadRemoteTimeout())
		if err != nil {
			if atomic.LoadUint32(&s.status) == closedFlag {
				return
			} else {
				logger.Errorf("[%s] read data failed. %s", s.id, err)
				return
			}
		}

		switch tag {
		case consts.Lsp2lcpTagTaskResp:
			var taskresp = new(proto.TaskResp)
			err := gproto.Unmarshal(proto.SplitPayload(payload)[0], taskresp)
			if err != nil {
				logger.Errorf("[%s] unmarshal data failed. %s", s.id, err)
				return
			}

			logger.Debugf("[%s] receive data. msgid:%d", s.id, taskresp.GetId())

			s.handleTask(taskresp)

		default:
			logger.Errorf("[%s] can't handle this tag. %d", s.id, tag)
		}
	}
}

func (s *server) handleTask(taskresp *proto.TaskResp) {
	s.Lock()
	defer s.Unlock()

	if cimsg, ok := s.reqMap[taskresp.GetId()]; ok {

		var comsg *clientOutMsg

		switch taskresp.GetStatus() {
		case consts.ServerAgentOK:
			comsg = &clientOutMsg{tag: consts.Lcp2lcTaskRespOK, data: taskresp.GetData()}

		case consts.ServerAgentBusy:
			comsg = &clientOutMsg{tag: consts.Lcp2lcTaskRespErrTry, data: []byte("server agent is busy")}

		case consts.ServerAgentWriteBusinessServerError:
			comsg = &clientOutMsg{tag: consts.Lcp2lcTaskRespErrTry, data: []byte("server agent write business server error")}

		case consts.ServerAgentReadBusinessServerError:
			comsg = &clientOutMsg{tag: consts.Lcp2lcTaskRespErrDontTry, data: []byte("server agent read business server error")}

		case consts.ServerAgentWaitingTimeout:
			comsg = &clientOutMsg{tag: consts.Lcp2lcTaskRespErrTry, data: []byte("server agent waiting timeout")}

		case consts.ServerAgentInternalError:
			comsg = &clientOutMsg{tag: consts.Lcp2lcTaskRespErrDontTry, data: []byte("server agent internal error")}

		default:
			comsg = &clientOutMsg{tag: consts.Lcp2lcTaskRespErrDontTry, data: []byte(fmt.Sprintf("unrecognized server agent status. %d", taskresp.GetStatus()))}
		}

		select {
		case cimsg.retCh <- comsg:
		default:
			//理论上应该不会运行到此处，记录日志，直接丢弃数据
			logger.Errorf("[%s] write task failed, drop message, msgid:%d", s.id, taskresp.GetId())
		}

		delete(s.reqMap, taskresp.GetId())

	} else {
		//出现这种情况可能是延时太久消息映射被删除了
		logger.Errorf("[%s] can't find client in reqMap:%d", s.id, taskresp.GetId())
	}

	if taskresp.GetId()%10000 == 0 {
		for id, cimsg := range s.reqMap {
			if cimsg != nil {
				if sub := time.Since(cimsg.createdAt); sub > time.Minute*3 {
					logger.Debugf("[%s] expire element in reqMap, msgid:%d", s.id, id)
					delete(s.reqMap, id)
				}
			}
		}
	}
}

func (s *server) close() {
	s.Lock()
	defer s.Unlock()

	if s.status != closedFlag {
		s.status = closedFlag

		if s.parser != nil {
			s.parser.Close()
		}

		if len(s.reqMap) == 0 {
			logger.Debugf("[%s] close", s.id)
		} else {
			var str = fmt.Sprintf("[%s] close, drop msg below:", s.id)
			for _, msg := range s.reqMap {
				if msg != nil {
					str += fmt.Sprintf("msgid:%d, data:%s \n", msg.msgid, string(msg.data))
				}
			}
			logger.Error(str)
		}
	}
}

func (s *server) generateMsgID() uint64 {
	s.Lock()
	s.msgid++
	id := s.msgid
	s.Unlock()
	return id
}
